package com.mqtt.listener;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.mqtt.service.MqttService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;

/**
 * Kafka消息监听器
 */
@Slf4j
@Component
public class KafkaMessageListener {
    
    @Autowired
    private MqttService mqttService;
    
    /**
     * 监听设备命令topic
     */
    @KafkaListener(topics = "device-command", groupId = "mqtt-service-group")
    public void handleDeviceCommand(ConsumerRecord<String, String> record, Acknowledgment ack) {
        try {
            log.debug("Received command from Kafka: key={}, value={}", 
                    record.key(), record.value());
            
            // 解析命令
            JSONObject command = JSON.parseObject(record.value());
            String deviceId = command.getString("deviceId");
            String action = command.getString("action");
            JSONObject params = command.getJSONObject("params");
            
            // 发送命令到设备
            Map<String, Object> commandParams = new HashMap<>();
            if (params != null) {
                commandParams.putAll(params.getInnerMap());
            }
            
            boolean sent = mqttService.sendCommand(deviceId, action, commandParams);
            
            if (sent) {
                log.info("Command sent to device {}: {}", deviceId, action);
            } else {
                log.error("Failed to send command to device {}: {}", deviceId, action);
            }
            
            // 确认消息
            ack.acknowledge();
            
        } catch (Exception e) {
            log.error("Error handling device command from Kafka", e);
            // 可以选择不确认，让消息重新消费
        }
    }
    
    /**
     * 监听设备配置更新
     */
    @KafkaListener(topics = "device-config-update", groupId = "mqtt-service-group")
    public void handleConfigUpdate(ConsumerRecord<String, String> record, Acknowledgment ack) {
        try {
            log.debug("Received config update from Kafka: {}", record.value());
            
            JSONObject config = JSON.parseObject(record.value());
            String deviceId = config.getString("deviceId");
            JSONObject configData = config.getJSONObject("config");
            
            // 构建配置更新消息
            Map<String, Object> updateMessage = new HashMap<>();
            updateMessage.put("type", "config_update");
            updateMessage.put("config", configData);
            updateMessage.put("timestamp", System.currentTimeMillis());
            
            // 发送到设备
            String topic = "device/" + deviceId + "/config";
            mqttService.sendToTopic(topic, JSON.toJSONString(updateMessage));
            
            log.info("Config update sent to device: {}", deviceId);
            
            // 确认消息
            ack.acknowledge();
            
        } catch (Exception e) {
            log.error("Error handling config update from Kafka", e);
        }
    }
    
    /**
     * 监听批量操作
     */
    @KafkaListener(topics = "device-batch-operation", groupId = "mqtt-service-group")
    public void handleBatchOperation(ConsumerRecord<String, String> record, Acknowledgment ack) {
        try {
            log.info("Received batch operation from Kafka");
            
            JSONObject operation = JSON.parseObject(record.value());
            String operationType = operation.getString("type");
            JSONObject devices = operation.getJSONObject("devices");
            JSONObject params = operation.getJSONObject("params");
            
            // 根据操作类型处理
            switch (operationType) {
                case "firmware_update":
                    handleFirmwareUpdate(devices, params);
                    break;
                case "reboot":
                    handleBatchReboot(devices);
                    break;
                case "config_sync":
                    handleConfigSync(devices, params);
                    break;
                default:
                    log.warn("Unknown batch operation type: {}", operationType);
            }
            
            // 确认消息
            ack.acknowledge();
            
        } catch (Exception e) {
            log.error("Error handling batch operation from Kafka", e);
        }
    }
    
    /**
     * 处理固件更新
     */
    private void handleFirmwareUpdate(JSONObject devices, JSONObject params) {
        log.info("Processing firmware update for {} devices", devices.size());
        
        for (String deviceId : devices.keySet()) {
            try {
                Map<String, Object> updateCommand = new HashMap<>();
                updateCommand.put("url", params.getString("url"));
                updateCommand.put("version", params.getString("version"));
                updateCommand.put("checksum", params.getString("checksum"));
                
                mqttService.sendCommand(deviceId, "firmware_update", updateCommand);
                
                // 添加延迟，避免消息风暴
                Thread.sleep(100);
                
            } catch (Exception e) {
                log.error("Failed to send firmware update to device: {}", deviceId, e);
            }
        }
    }
    
    /**
     * 处理批量重启
     */
    private void handleBatchReboot(JSONObject devices) {
        log.info("Processing batch reboot for {} devices", devices.size());
        
        for (String deviceId : devices.keySet()) {
            try {
                mqttService.sendCommand(deviceId, "reboot", new HashMap<>());
                
                // 添加延迟
                Thread.sleep(50);
                
            } catch (Exception e) {
                log.error("Failed to send reboot command to device: {}", deviceId, e);
            }
        }
    }
    
    /**
     * 处理配置同步
     */
    private void handleConfigSync(JSONObject devices, JSONObject params) {
        log.info("Processing config sync for {} devices", devices.size());
        
        for (String deviceId : devices.keySet()) {
            try {
                Map<String, Object> syncCommand = new HashMap<>();
                syncCommand.put("config", params);
                
                mqttService.sendCommand(deviceId, "config_sync", syncCommand);
                
                // 添加延迟
                Thread.sleep(100);
                
            } catch (Exception e) {
                log.error("Failed to send config sync to device: {}", deviceId, e);
            }
        }
    }
}